package org.luckyjourney.util;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import org.luckyjourney.constant.RedisConstant;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * @creator: liRong
 * @date: 2025/6/24 22:38
 */
@Component
public class VideoDeduplicationUtil {
    // 每个用户一个布隆过滤器
    private final Map<Long , BloomFilter<String>> userFilters = new HashMap<>();


    StringRedisTemplate redisTemplate;

    public VideoDeduplicationUtil(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    // 去重主方法
    public Set<Long> deduplicate(Set<Long> videoIds , Long userId) {
        // 1.获取后创建当前用户的布隆过滤器
        BloomFilter<String> bloomFilter = getOrCreateFilter(userId);

        // 2.准备两个集合：过滤后返回的ID集合 和 需要二次检查的ID
        Set<Long> resultIds = new HashSet<>();
        List<Long> checkIds = new ArrayList<>();

        // 3.布隆过滤器筛选
        for (Long videoId : videoIds) {
            String key = buildKey(videoId , userId);
            if(bloomFilter.mightContain(key)) {
                // 可能存在 -- 需要二次检查 , 暂时不添加到res
                checkIds.add(videoId);
            } else {
                // 一定不存在 , 直接添加到res
                resultIds.add(videoId);
            }
        }

        // 4.对需要检查的 ID进行二次检查
        if(!checkIds.isEmpty()) {
            List<Object> checkRedisRes = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                for (Long videoId : checkIds) {
                    String key = RedisConstant.HISTORY_VIDEO + videoId + ":" + userId;
                    connection.get(key.getBytes());
                }
                return null;
            });

            for (int i = 0; i < checkIds.size(); i++) {
                if(ObjectUtils.isEmpty(checkRedisRes.get(i))) {
                    // Redis中不存在 这个历史记录，则保留到res
                    resultIds.add(checkIds.get(i));
                }
            }
        }

        // 5.将新发现的ID 添加到过滤器中
        for(Long newId : resultIds) {
            bloomFilter.put(buildKey(newId , userId));
        }

        return resultIds;
    }

    private String buildKey(Long videoId, Long userId) {
        return userId + ":" + videoId;
    }

    private BloomFilter<String> getOrCreateFilter(Long userId) {
        return userFilters.computeIfAbsent(userId , id ->
            BloomFilter.create(
                    Funnels.stringFunnel(StandardCharsets.UTF_8),
                    10000, // 每个用户最多10000个视频
                    0.01 // 1%的误判率
            )
        );
    }
}
